package com.xctech.yace.thread;

import com.xctech.yace.single.RecivedDataManger;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerThread implements Runnable{

    public static AtomicInteger count = new AtomicInteger(0);

    private KafkaConsumer<String,String> consumer;

    private ConsumerRecords<String, String> msgs;

    private final String[] topics;


    public ConsumerThread(KafkaConsumer consumer,String[] topicName) {
        this.consumer = consumer;
        this.topics = topicName;
        this.consumer.subscribe(Arrays.asList(topicName));
    }


    @Override
    public void run() {
        System.out.println("-----------开始消费-------------");
        try{
            for(;;){
                msgs = consumer.poll(1);
                RecivedDataManger<ConsumerRecord<String,String>> instance = RecivedDataManger.getInstance();
                if(null != msgs && msgs.count() > 0){
                    for (ConsumerRecord<String,String> record:msgs){
//                        System.out.println("=======receive: key = " + record.key() +
//                                ", value = " + record.value()+" offset==="+record.offset());
                        //2.计算间隔
                        System.out.println("=======receive: key = "+record.value());
//                        Long endTime = TimeUtil.getNaTime();
//                        Long startTime = Long.parseLong(record.key());
//                        System.out.println("=======receive: time = "+"开始时间："+startTime+"|"+"结束时间："+endTime+"耗时："+TimeUtil.calInterval(startTime,endTime));
                        System.out.println("=======receive: time 耗时："+ record.key()+"ms");

                        //存数据到集合中
                        instance.push(record);
                    }
                }else {
                    Thread.sleep(10);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        } finally {
//            consumer.close();
        }
    }





}
